package com.usfot;

import org.apache.zookeeper.KeeperException;

/**
 * 测试基于zookeeper的生产者和消费者队列
 * 测试过程：
 * 一个线程充当生产者，一个线程充当消费者
 * Created by liyanxin on 2015/3/17.
 */
public class QueueTest {

    public static void main(String args[]) {
        Producer producer = new Producer(new Queue("/app1"));
        //生产者线程启动
        producer.start();

        // 启动多个线程作为消费者，但在当前的分布式环境中，消费者之间会竞争某些资源，也是说任务不是
        // 同步进行的，导致一个消费者想要得到某个节点的数据时，而这个节点却被另一个消费者删除了，
        // 导致KeeperException$NoNodeException
//        for (int i = 0; i < 100; i++) {
//            Consumer consumer = new Consumer(new Queue("/app1"));
//            consumer.start();
//        }

        // 消费者线程
        Consumer consumer = new Consumer(new Queue("/app1"));
        consumer.start();
    }
}

/**
 * 生产者线程
 */
class Producer extends Thread {

    private Queue queue;

    public Producer(Queue queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        // 该生产者线程创建1000个znode
        for (int i = 0; i < 100; i++) {
            try {
                queue.produce(i);
                Thread.sleep(1000);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

/**
 * 消费者线程
 */
class Consumer extends Thread {
    private Queue queue;

    public Consumer(Queue queue) {
        this.queue = queue;
    }

    // 消费者线程
    // 不断的从zk服务器中读取znode，进行操作。
    @Override
    public void run() {
        try {
            while (true) {
                int retvalue = queue.consume();
                System.out.println("thread_name=" +
                        Thread.currentThread().getName() + ":" + retvalue);
                Thread.sleep(1000);
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}